package ldh.im.rpc.connection.handle;

import ldh.im.rpc.connection.Receive;
import ldh.im.rpc.connection.RpcCallback;
import ldh.im.rpc.exception.ExceptionResp;
import ldh.im.rpc.exception.RpcErrorCode;
import ldh.im.rpc.exception.RpcException;
import ldh.im.rpc.exception.RpcResultErrorCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

public class DefaultMessageCallback implements MessageCallback {

    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultMessageCallback.class);

    private Map<String, InitData> callbackMap = new ConcurrentHashMap<>(); // 回调函数

    public DefaultMessageCallback() {
        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                return thread;
            }
        });

        executorService.scheduleAtFixedRate(()->{
            cleanTimeoutCallback();
        }, 10, 60, TimeUnit.SECONDS);
    }

    @Override
    public void addRpcFuture(String seq, CompletableFuture future) {
        Receive<Object> receive = buildCallback(future);
        addReceive(seq, receive);
    }

    @Override
    public void addRpcFuture(String seq, RpcCallback rpcCallback) {
        Receive<Object> receive = buildCallback(rpcCallback);
        addReceive(seq, receive);
    }

    private void addReceive(String seq, Receive<Object> receive) {
        callbackMap.put(seq, new InitData(receive));

        int size = callbackMap.size();
        LOGGER.info("message callback size : {}, {}", size, this);

        if (size > 100) {
            LOGGER.warn("========================message callback size too long {}=========", size);
        }
    }

    private Receive<Object> buildCallback(CompletableFuture future) {
        Receive<Object> consumer = (to)->{
            if (to instanceof ExceptionResp) {
                ExceptionResp exceptionResp = (ExceptionResp) to;
                future.completeExceptionally(new RpcException(new RpcResultErrorCode(exceptionResp.getErrorStatus(),
                    exceptionResp.getErrorMsg())));
                return;
            }

            future.complete(to);
        };
        return consumer;
    }

    private Receive<Object> buildCallback(RpcCallback rpcCallback) {
        Receive<Object> consumer = (to)->{
            if (to instanceof ExceptionResp) {
                ExceptionResp exceptionResp = (ExceptionResp) to;
                rpcCallback.callback(null, new RpcException(RpcErrorCode.Result_error, exceptionResp.getErrorStatus() + ":" + exceptionResp.getErrorMsg()));
                return;
            }

            rpcCallback.callback(to, null);
        };
        return consumer;
    }

    private void cleanTimeoutCallback() {
        LocalDateTime now = LocalDateTime.now();
        Set<String> removeKeys = new HashSet();
        callbackMap.forEach((key, value)->{
            if (value.getCreateTime().plusSeconds(2 * 60 * 60).isBefore(now)) {
                removeKeys.add(key);
            }
        });

        removeKeys.forEach(key->{
            callbackMap.remove(key);
        });

//        LOGGER.info("message callback size (cleaned): {}, {}", callbackMap.size(), JsonUtil.toJson(callbackMap.keySet()));
    }

    private class InitData {
        private LocalDateTime createTime;
        private Receive<Object> receive;

        public InitData(Receive<Object> receive) {
            this.createTime = LocalDateTime.now();
            this.receive = receive;
        }

        public LocalDateTime getCreateTime() {
            return createTime;
        }

        public Receive<Object> getReceive() {
            return receive;
        }
    }
}
